package com.ahaha

import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SparkCluster {
  def main(args: Array[String]): Unit = {
    if (args == null || args.length != 1){
      println(
        """
          |啥玩意啊
          |""".stripMargin)
      System.exit(-1)
    }
    val Array(inputpath) = args
    
    Logger.getLogger("org.spark_project").setLevel(Level.INFO)
    Logger.getLogger("org.apache.spark").setLevel(Level.INFO)

    val conf = new SparkConf()
      //.setMaster("local[*]")
      .setAppName(s"${SparkCluster.getClass.getSimpleName}")

    val sc = new SparkContext(conf)
    val lines:RDD[String] = sc.textFile(inputpath)
    println("partitions:" + lines.getNumPartitions)
    val words:RDD[String] = lines.flatMap(line => line.split("\\s+"))
    val pairs:RDD[(String,Int)] = words.map(word => (word, 1))
    val ret:RDD[(String,Int)] = pairs.reduceByKey((v1, v2) => v1 + v2)
    ret.foreach(t=>println(t))

    sc.stop()
  }

}
